#include "../lib/plumber.hpp"
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <sys/ioctl.h>
#include <stdio.h>

Plumber::Plumber(int port, RoutingIndex r)
{
    port_ = port;
    index_ = r;
}

bool Plumber::setUpServerSocket_()
{
    struct sockaddr_in serv_addr;
    masterSocket_ = socket(AF_INET, SOCK_STREAM, 0);
    if (masterSocket_ < 0)
        return false;
    bzero((char *) &serv_addr, sizeof(serv_addr));
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_addr.s_addr = INADDR_ANY;
    serv_addr.sin_port = htons(port_);
    int yes = 1;

    setsockopt(masterSocket_, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int));

    if (bind(masterSocket_, (struct sockaddr *) &serv_addr, sizeof(serv_addr)) < 0)
        return false;
    listen(masterSocket_,15);
    return true;
}

#include <iostream>

using namespace std;

bool Plumber::start()
{
    struct epoll_event event;
    if(!setUpServerSocket_())
        return false;
    efd_ = epoll_create1(0);
    event.events = EPOLLIN;
    event.data.ptr = new char[14];
    ::sprintf((char*)event.data.ptr, "$MasterSocket");
    names_.push_back((char*)event.data.ptr);
    if(epoll_ctl (efd_, EPOLL_CTL_ADD, masterSocket_, &event) == -1)
    {
        ::close(efd_);
        ::close(masterSocket_);
        delete[] (char*)event.data.ptr;
        names_.clear();
        return false;
    }
    t_ = new boost::thread(&Plumber::workLoop_, this);
    return true;
}

bool Plumber::stop()
{
    t_->interrupt();
    t_->join();
    close(masterSocket_);
    return true;
}

void Plumber::workLoop_()
{
    try
    {
        struct epoll_event events[MAX_EVENTS];
        while(true)
        {
            boost::this_thread::interruption_point();
            struct timeval now;
            gettimeofday(&(now), NULL);

            int eventNum = epoll_wait(efd_, events, MAX_EVENTS, 10);
            if(eventNum < 0)
                return;
            for(int i = 0; i< eventNum; i++)
            {
                std::string name((char*)events[i].data.ptr);
                if(name == "$MasterSocket")
                    handleAccept_();
                else
                    handleRequest_(name, now);
            }
            updateQueries_(now);

        }
    }
    catch(boost::thread_interrupted e)
    {

    }
}

void Plumber::handleAccept_()
{
    int client;
    struct sockaddr_in cliAddr;
    socklen_t cliLen;
    struct epoll_event event;

    cliLen = sizeof(cliAddr);
    client = ::accept(masterSocket_, (struct sockaddr *) &cliAddr, &cliLen);
    if(client < 0)
        return;
    spointer sp = spointer(new Session(client));
    if(!sp->isOk())
    {
        return;
    }
    clients_[sp->getName()] = sp;
    int nameLen = sp->getName().length();
    char * name = new char[nameLen];
    sprintf(name, "%s", sp->getName().c_str());
    event.data.ptr = name;
    names_.push_back(name);
    event.events = EPOLLIN;
    epoll_ctl (efd_, EPOLL_CTL_ADD, client, &event);
}

void Plumber::handleRequest_(std::string name, struct timeval now)
{
    if(clients_.find(name) == clients_.end())
        return;
    spointer client = clients_[name];

    CmdHeader header = client->readHeader();
    std::vector <std::string> args;
    if(!client->isOk())
    {
        return;
    }
    if(header.payloadLen > 0)
        args = client->readArgs(header.payloadLen);
    switch (header.cmd)
    {
    case HEADER_SET_COMMAND:
    {
        std::string key = args[0];
        std::string value  = args[1];
        if(queries_.find(key) == queries_.end())
        {
            now.tv_sec = header.tv_sec;
            now.tv_usec = header.tv_usec;
            Query q(key, index_, now);
            queries_[key] = q;
        }
        queries_[key].publishers[name] = value;
    }
        break;
    case HEADER_RECIEVED_COMMAND:
    {
        std::string key = args[0];
        if(queries_.find(key) == queries_.end())
        {
            break;
        }
        struct timeval tv;
        tv.tv_sec = header.tv_sec;
        tv.tv_usec = header.tv_usec;
        queries_[key].subscribers[name] = tv;

    }
        break;
    case HEADER_SUBSCRIBE_READER:
    {
        std::string key = args[0];
        std::vector <std::string > subscribers = index_.getSubscriber(key);
        if(std::find(subscribers.begin(), subscribers.end(),name) == subscribers.end())
        {
            index_.addSubscriber(key, name);
        }
    }
        break;
    case HEADER_SUBSCRIBE_WRITER:
    {
        std::string key = args[0];
        std::vector <std::string > publishers = index_.getPublisher(key);
        if(std::find(publishers.begin(), publishers.end(),name) == publishers.end())
        {
            index_.addPublisher(key, name);
        }

    }
        break;
    default:
        cout << "Bad Header " << header.cmd << endl;
        break;

    }
}

void Plumber::updateQueries_(struct timeval now)
{
    std::map<std::string, Query>::iterator iter;
    std::vector<std::string> queriesToErase;
    for(iter =  queries_.begin(); iter != queries_.end(); ++iter)
    {
        std::string key = iter->first;
        if(iter->second.state == Query::WINDOW)
        {
            if(iter->second.windowFrom(now) >= WINDOWS_SIZE)
            {
                std::string value = pickSetValue(iter->second);
                CmdHeader h;
                h.cmd = HEADER_SET_COMMAND;
                h.tv_sec = now.tv_sec;
                h.tv_usec = now.tv_usec;
                std::vector<std::string> args;
                args.push_back(key);
                args.push_back(value);
                sendPackets(h, args, index_.getSubscriber(key));
                iter->second.state = Query::ACK;
            }

        }
        if(iter->second.state == Query::ACK)
        {
            if(iter->second.allAckRecieved())
            {
                iter->second.state = Query::RETURN;
            }
        }
        if(iter->second.state == Query::RETURN)
        {
            struct timeval latestAck = iter->second.latestAck();
            queriesToErase.push_back(iter->first);
            CmdHeader h;
            h.cmd = HEADER_SET_NO_ACK_COMMAND;
            h.tv_sec = latestAck.tv_sec;
            h.tv_usec = latestAck.tv_usec;
            std::vector<std::string> args;
            std::string value = pickSetValue(iter->second);
            args.push_back(key);
            args.push_back(value);
            sendPackets(h, args, index_.getPublisher(key));

        }
    }
    for(unsigned int i = 0; i < queriesToErase.size(); i++)
    {
        queries_.erase(queriesToErase[i]);
    }
}

bool Plumber::sendPackets(CmdHeader header, std::vector<std::string> args, std::vector<std::string> hosts)
{
    bool ret = true;
    for(unsigned int i = 0 ; i< hosts.size(); i++)
    {
        if(!clients_[hosts[i]]->write(header, args))
            ret = false;
    }
    return ret;
}

static bool compMap(const std::pair<std::string,unsigned int> &pair1, const std::pair<std::string,unsigned int> &pair2)
{
    return pair1.second < pair2.second;
}

std::string Plumber::pickSetValue(Query q)
{
    std::map<std::string, std::string>::iterator iter;
    std::map<std::string, unsigned int> setMap;
    for(iter =  q.publishers.begin(); iter != q.publishers.end(); ++iter)
    {
        ++setMap[iter->second];
    }
    return std::max_element(setMap.begin(), setMap.end(), compMap)->first;
}
